package com.ljm.reactor.create;

import reactor.core.publisher.Flux;
import reactor.util.function.Tuple2;
import reactor.util.function.Tuples;

import java.util.Comparator;
import java.util.List;

/**
 * @author 李佳明 https://github.com/pkpk1234
 * @date 2018-04-16
 *
 * WebFlux中的Sinks是一个重要的组件，主要用于管理消息流和实现复杂的消息处理逻辑。
 * Sinks是WebFlux中用于管理消息流的一个关键组件，它提供了一种机制来收集和分发消息。Sinks可以处理多种类型的消息流，包括单播（Mono）和多播（Flux）。Sinks的主要功能包括：
 * ‌管理消息流‌：通过Sinks，可以创建和管理消息流，实现消息的收集和分发。
 * ‌订阅和发送消息‌：使用Sinks可以方便地订阅和发送消息，支持分组发送和群发功能。
 * ‌缓冲和回压‌：Sinks提供了缓冲和流量控制的功能，确保在高并发情况下不会丢失消息
 */
public class FluxBridge {

    private static MyEventProcessor myEventProcessor = new ScheduledSingleListenerEventProcessor();

    public static void main(String[] args) {
        Flux.create(sink -> {
            myEventProcessor.register(
                    new MyEventListener<String>() {
                        public void onEvents(List<String> chunk) {
                            for (String s : chunk) {
                                if ("end".equalsIgnoreCase(s)) {
                                    sink.complete();
                                    myEventProcessor.shutdown();
                                } else {
                                    sink.next(s);
                                }

                            }
                        }
                        public void processComplete() {
                            sink.complete();
                        }
                    });
        }).log().subscribe(System.out::println);
        myEventProcessor.fireEvents("abc", "efg", "123", "456", "end");

/* 以下是模拟调试 jetlinks中网关分页查询 的逻辑
        //响应式编程里，竟然是先执行下边这一段
        Flux.just("str1")
                .map(item -> {return   item.length();})
                .log()
                .subscribe(System.out::println); //4
        Flux.just("str1" , "str")
                .map(item -> {return   item.length();})
                .log()
                .subscribe(System.out::println); //[4,3] 分别打印的

        Flux.just("str1" , "str")
                .map(item -> {return   item.length();})
                .collectList()
                .log()
                .subscribe(System.out::println); //[4,3] 作为list打印的

        Flux.just("str1")
                .map(item -> {return   item.length();})
                .index()
                .log()
                .subscribe(System.out::println); //[0,4]


        Flux.just("str1" , "str")
                .map(item -> {return   item.length();})
                .index()
                .log()
                .subscribe(System.out::println);//[0,4] [1,3]

//        Flux.just("str1" , "str")
//                .map(item -> {return   item.length();})
//                .index()
//                .flatMap(t2->{ return  Flux.just(t2.getT2());})
//                .log()
//                .subscribe(System.out::println);//4 3

        // 第一行是初始化一个字符串的flux流；
        // 然后通过map()转换，对每个元素运算，将其转为了int flux；
        // index() 将每个元素转化为了 tuple2[nindex, 元素int]
        //

//        Flux.just("str1" , "str")
//                .map(item -> {return   item.length();})
//                .index()
//                .flatMap(t2-> Flux.just(t2.getT2()).map(detail->Tuples.of(t2.getT2(), detail)))
//                .log()
//                .subscribe(System.out::println);//4 3

        // -> 4、3、6、11
        // ->index [0,4] [1,3] [2,6] [3,11]
        // ->flatMap [1,3]->3->[3,3] [0,4]->4->[4,4] [2,6]->6->[6,6] [3,11]->11->[11,11]
        // ->sort [3,3] [4,4] [6,6] [11,11]
        // ->map 3 4 6 11
        // ->collectList 生成list了; 如果没有collectList，就是打印出（单独处理完）每个元素，而不是list数组
        Flux.just("str1" , "str", "string", "string_help")
                .map(item -> {return   item.length();})
                .index()
                .flatMap(t2-> Flux.just(t2.getT2()).map(detail->Tuples.of(t2.getT2(), detail)))
                .sort(Comparator.comparingLong(Tuple2::getT1))
                .map(Tuple2::getT2)
                .collectList()
                .log()
                .subscribe(System.out::println);//4 3


*/

        System.out.println("main thread exit");
    }
}

